package com.lizhiyu.flink.demo1;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


/***
 * Flink 流处理
 * 简单设置DataStream返回格式数据源，将数据源通过flapMap转换后打印
 */
public class A1_Flink_Demo1 {

    public static void main(String[] args) throws Exception {
        //流处理用 StreamExecutionEnvironment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

        //并行执行数量
        env.setParallelism(2);

        //设置source
        DataStream<String> stringDataStream = env.fromElements("haha", "hehe,xixi", "enen,oo");

        //打印source
        stringDataStream.print();

        //通过flatMap 进行转换下
        DataStream<String> flatMapDataStream = stringDataStream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String s, Collector<String> collector) throws Exception {
                String[] split = s.split(",");
                for (String word : split) {
                    collector.collect(word);
                }
            }
        });
        //打印
        flatMapDataStream.print(Thread.currentThread().getName()+":");

        env.execute("JOB 1");

    }
}
